/*
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this file,
 * You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 * Portions created by the Initial Developer are
 * Copyright (C) 2012 the Initial Developer.
 * All Rights Reserved.
 *
 * The Original Code is this.
 *
 * The Initial Developer of the Original Code is the author below.
 */
/**
 * @file MessageQue.cpp
 * @brief auto generated.
 * @author K.Ozaki
 * @date 2011/03/11
 */

#include "semi/posix/MessageQue.h"

namespace semi {

namespace posix {

SignalAction* MessageQue::sig_action = new SignalAction(SIGUSR1);
// int MessageQue::SigNum = SIGRTMIN;
std::string MessageQue::que_id_file_name = "semi.CurrentQueNames";

std::string MessageQue::terminate_message = "terminate_message";
std::string MessageQue::empty_message = "empty_message";
std::string MessageQue::error_message = "error_message";

//MessageQue::MessageQue() {
//}

MessageQue::MessageQue(std::string quename) {
    init(quename);
}
MessageQue::~MessageQue() {
    //    this->sem.terminate();
    //    this->mqpl.end();
}

void MessageQue::init(std::string quename) {

    this->QueName = this->createQuename(quename);
    this->mqpl.init(que_id_file_name, sizeof(MessageQue::QuePool));
    //    this->sem.init(que_id_file_name);
}

std::string MessageQue::createQuename(std::string quename) {

    std::string name;
    name = semi::util::StringUtil::replace(quename, "/", "");
    name = "/" + name;

    int siz = 32;
    if (siz < name.size())
        name = name.substr(0, siz);
    return name;
}

std::string MessageQue::getQueName() {
    return QueName;
}

int MessageQue::addQueCount(bool send) {

    if (0 == this->QueName.length())
        return -1;

    MessageQue::QuePool* qpol = (MessageQue::QuePool*) mqpl.get();
    for (int idx = 0; idx < (sizeof(MessageQue::QuePool)
            / sizeof(MessageQue::QueRecord)); idx++) {
        std::string tmp(qpol->ques[idx].queName);
        if (0 < tmp.length() && tmp == this->QueName) {
            if (qpol->ques[idx].send_count < 0)
                qpol->ques[idx].send_count = 0;
            if (qpol->ques[idx].wait_count < 0)
                qpol->ques[idx].wait_count = 0;

            if (send)
                qpol->ques[idx].send_count = qpol->ques[idx].send_count + 1;
            // notify
            else
                qpol->ques[idx].wait_count = qpol->ques[idx].wait_count + 1;

            mqpl.end();
            return idx;
        }
    }
    for (int idx = 0; idx < (sizeof(MessageQue::QuePool)
            / sizeof(MessageQue::QueRecord)); idx++) {
        std::string tmp(qpol->ques[idx].queName);
        // if (tmp.length() < 0 || qpol->ques[idx].send_count <= 0) {
        if (0 == tmp.length()) {
            memset(qpol->ques[idx].queName, 0, sizeof(qpol->ques[idx].queName));
            qpol->ques[idx].send_count = 0;
            qpol->ques[idx].wait_count = 0;

            strcpy(qpol->ques[idx].queName, this->QueName.c_str());
            if (send)
                qpol->ques[idx].send_count = 1;
            // notify
            else
                qpol->ques[idx].wait_count = 1;

            mqpl.end();
            return idx;
        }
    }
    mqpl.end();
    // over flow
    std::cout << std::endl << "    MessageQue::addQueCount over flow!! ";
    if (send)
        std::cout << "send:";
    else
        std::cout << "wait:";
    std::cout << QueName << std::endl;

    return -1;
}

int MessageQue::removeQueCount(bool send) {
    MessageQue::QuePool* qpol = (MessageQue::QuePool*) mqpl.get();
    for (int idx = 0; idx < (sizeof(MessageQue::QuePool)
            / sizeof(MessageQue::QueRecord)); idx++) {
        std::string tmp(qpol->ques[idx].queName);
        if (0 < tmp.length() && tmp == this->QueName) {

            if (send && 0 < qpol->ques[idx].send_count)
                qpol->ques[idx].send_count = qpol->ques[idx].send_count - 1;
            else if (!send && 0 < qpol->ques[idx].wait_count)
                qpol->ques[idx].wait_count = qpol->ques[idx].wait_count - 1;
            mqpl.end();
            return idx;
        }
    }
    mqpl.end();
    return -1;
}

int MessageQue::send(std::string message, unsigned int prio) {

    //    std::cout << "    MessageQue::send start " << QueName << " message:"
    //            << message << std::endl;

    //    mqd_t mqd = mq_open(this->QueName.c_str(), O_RDWR | O_NONBLOCK | O_CREAT,
    //            (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH), NULL);
    mqd_t mqd = mq_open(this->QueName.c_str(), O_WRONLY | O_NONBLOCK | O_CREAT,
            (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH), NULL);

    if (mqd <= 0) {
        // retry
        mqd = mq_open(this->QueName.c_str(), O_WRONLY | O_NONBLOCK | O_CREAT,
                (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH), NULL);

        if (mqd <= 0) {
            std::stringstream ss;
            ss << "  MessageQue::send mq_open error que name:"
                    << QueName.c_str() << " mqd:" << mqd;
            perror(ss.str().c_str());
            //            throw std::logic_error(
            //                    "MessageQue::receive mq_open error que name:"
            //                            + this->QueName);
            return mqd; // fail.
        }
    }

    int siz = message.length();
    char *ptr = (char*) calloc(siz + 1, sizeof(char));
    strncpy(ptr, message.c_str(), siz);

    //    this->sem.wait();
    this->addQueCount(true); // send +1
    int ret = mq_send(mqd, ptr, siz, prio);
    //    this->sem.post();
    //
    //    std::cout << "    MessageQue::sended " << QueName << " ret:" << ret
    //            << " mqd:" << mqd << std::endl;

    free(ptr);

    mq_close(mqd);

    return ret;

}

std::string MessageQue::receive(unsigned int prio) {

    mqd_t mqd = mq_open(this->QueName.c_str(), O_RDONLY | O_NONBLOCK | O_CREAT,
            (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH), NULL);

    if (mqd <= 0) {
        // retry
        mqd = mq_open(this->QueName.c_str(), O_RDONLY | O_NONBLOCK | O_CREAT,
                (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH), NULL);
        if (mqd <= 0) {
            std::stringstream ss;
            ss << "  MessageQue::receive mq_open error que name:"
                    << QueName.c_str() << " mqd:" << mqd;
            perror(ss.str().c_str());
            //            throw std::logic_error(
            //                    "MessageQue::receive mq_open error que name:"
            //                            + this->QueName);
            //            mq_close(mqd);
            return ""; // fail.
        }
    }

    struct mq_attr attr;
    mq_getattr(mqd, &attr);

    char* buff = (char*) malloc(attr.mq_msgsize);
    //    char* buff =0;

    //    this->sem.wait();
    // ssize_t mq_receive(mqd_t, char*, size_t, unsigned int*)'
    ssize_t n = mq_receive(mqd, buff, attr.mq_msgsize, &prio);
    //    std::cout << "    MessageQue::receive " << QueName << " mq_receive " << n
    //            << std::endl;

    //    if (errno != EAGAIN) {
    //        perror("mq_receive");
    //    }
    if (n < 0) {
        //        std::cout << "    MessageQue::receive " << QueName
        //                << " mq_receive fail mqd:" << mqd << " len:" << n << std::endl;
        free(buff);
        //        this->sem.post();
        mq_close(mqd);
        return ""; // fail.
    }

    // ok.
    this->removeQueCount(true); // send -1
    //    this->sem.post();

    if (0 == n) {
        free(buff);
        mq_close(mqd);
        return MessageQue::empty_message;
    }

    std::string ret(buff, n);
    free(buff);

    mq_close(mqd);
    //    std::cout << "    MessageQue::receive " << QueName << " mqd:" << mqd
    //            << " mq_receive " << ret.c_str() << std::endl;

    return ret;
}

std::string MessageQue::notify(unsigned int prio) {

    // already send?
    MessageQue::QuePool* qpol =
            (MessageQue::QuePool*) this->mqpl.get(PROT_READ);
    for (int i = 0; i < (sizeof(MessageQue::QuePool)
            / sizeof(MessageQue::QueRecord)); i++) {
        std::string tmp(qpol->ques[i].queName);
        if (0 < tmp.length() && tmp == this->QueName && 0
                < qpol->ques[i].send_count) {
            std::string ret = this->receive(prio);
            std::cout << "    MessageQue::mq_notify already sended receive "
                    << i << " " << ret << std::endl;
            if (0 < ret.length()) {
                this->mqpl.end();
                return ret;
            }
            break; // fail to receive
        }
    }
    //  this->mqpl.end();

    mqd_t mqd = mq_open(this->QueName.c_str(), O_RDWR | O_NONBLOCK | O_CREAT,
    //    mqd_t mqd = mq_open(this->QueName.c_str(), O_RDONLY | O_NONBLOCK | O_CREAT,
            (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH), NULL);
    if (mqd <= 0) {
        std::stringstream ss;
        ss << "  MessageQue::notify mq_open error que name:" << QueName.c_str()
                << " mqd:" << mqd;
        perror(ss.str().c_str());
        //            throw std::logic_error(
        //                    "MessageQue::receive mq_open error que name:"
        //                            + this->QueName);
        this->mqpl.end();
        return MessageQue::error_message;
    }

    sigevent sigev = MessageQue::sig_action->create();

    int i = mq_notify(mqd, &sigev);
    int idx = this->addQueCount(false); // wait +1

    // break control
    char flgChk = '1';
    //    memcpy(&qpol->flg, &flgChk, 1); // save to shared memory

    std::string ret = "";

    while ('1' == flgChk) { // wait signal

        //        std::cout << "    MessageQue::mq_notify sigwait... " << this->QueName
        //                << std::endl;

        // sleep(1);
        // sigsuspend(&zeromask);
        MessageQue::sig_action->wait(); // sigwaits (break for all que)
        std::cout << "    MessageQue signal break." << std::endl;

        this->mqpl.end();
        qpol = (MessageQue::QuePool*) this->mqpl.get(PROT_READ);
        //        memcpy(&flgChk, &qpol->ques[idx].flg, 1); // get from shared memory
        //        if ('0' == flgChk) // check break flg
        //            break;

        for (int i = 0; i < (sizeof(MessageQue::QuePool)
                / sizeof(MessageQue::QueRecord)); i++) {
            std::string tmp(qpol->ques[i].queName);
            if (0 < tmp.length() && tmp == this->QueName) {

                if (0 < qpol->ques[i].send_count) {
                    ret = this->receive(prio);
                    flgChk = '0'; // loop break!
                    break;
                } else if (qpol->ques[i].wait_count < 1) {
                    std::cout << "    MessageQue not wait!" << std::endl;
                    flgChk = '0'; // loop break!
                    break;
                }
            }
        }
        //  this->mqpl.end();
    }
    this->removeQueCount(false); // wait -1
    this->mqpl.end();
    mq_close(mqd);
    return ret;
}

int MessageQue::terminate() {

    // send all reset que wait_count
    // receive all reset que send_count
    MessageQue::QuePool* qpol = (MessageQue::QuePool*) this->mqpl.get();
    for (int i = 0; i < (sizeof(MessageQue::QuePool)
            / sizeof(MessageQue::QueRecord)); i++) {
        std::string tmp(qpol->ques[i].queName);
        if (0 < tmp.length() && tmp == this->QueName) {

            for (int j = 0; j < qpol->ques[i].wait_count; j++)
                this->send(terminate_message, 1);
            qpol->ques[i].wait_count = 0;
            sleep(1);
            for (int j = 0; j < qpol->ques[i].send_count; j++)
                this->receive(1);
            qpol->ques[i].send_count = 0;
            break;
        }
    }
    this->mqpl.end();

    return mq_unlink(QueName.c_str());
}

// for shut down.
void MessageQue::releaseAllMessage() {

    //    std::cout << "    MessageQue::terminateAll" << std::endl;

    SharedMemory crntmq(que_id_file_name, sizeof(MessageQue::QuePool));
    MessageQue::QuePool* qpol = (MessageQue::QuePool*) crntmq.get(PROT_READ);

    for (int i = 0; i < (sizeof(MessageQue::QuePool)
            / sizeof(MessageQue::QueRecord)); i++) {
        std::string nm(qpol->ques[i].queName);
        if (0 < nm.length()) {
            for (int j = 0; j < qpol->ques[i].wait_count; j++)
                MessageQue(nm).send(terminate_message, 5);
        }
    }

    for (int i = 0; i < (sizeof(MessageQue::QuePool)
            / sizeof(MessageQue::QueRecord)); i++) {
        std::string nm(qpol->ques[i].queName);
        if (0 < nm.length()) {
            for (int j = 0; j < qpol->ques[i].send_count; j++)
                MessageQue(nm).receive(5);
        }
    }
    crntmq.end();

    //    qpol = (MessageQue::QuePool*) crntmq.get();
    //    for (int i = 0; i < (sizeof(MessageQue::QuePool)
    //            / sizeof(MessageQue::QueRecord)); i++) {
    //        std::string nm(qpol->ques[i].queName);
    //        if (0 < nm.length()) {
    //            for (int j = 0; j < qpol->ques[i].send_count; j++)
    //                MessageQue(nm).receive(1);
    //        }
    //    }

    //    qpol = (MessageQue::QuePool*) crntmq.get();
    //    for (int i = 0; i < (sizeof(MessageQue::QuePool)
    //            / sizeof(MessageQue::QueRecord)); i++) {
    //        std::string nm(qpol->ques[i].queName);
    //        if (0 < nm.length()) {
    //            mq_unlink(nm.c_str());
    //        }
    //    }
    //    memset(qpol, 0, sizeof(MessageQue::QuePool));
    //    crntmq.end();
}

// for start up.
void MessageQue::clearCurrentQueNames() {
    SharedMemory crntmq(que_id_file_name, sizeof(MessageQue::QuePool));
    MessageQue::QuePool* qpol = (MessageQue::QuePool*) crntmq.get();

    qpol = (MessageQue::QuePool*) crntmq.get();
    for (int i = 0; i < (sizeof(MessageQue::QuePool)
            / sizeof(MessageQue::QueRecord)); i++) {
        std::string nm(qpol->ques[i].queName);
        if (0 < nm.length()) {
            mq_unlink(nm.c_str());
        }
    }

    memset(qpol, 0, sizeof(MessageQue::QuePool));
    crntmq.end();
}

std::string MessageQue::toStringCurrentQue(bool bl) {

    SharedMemory crntmq(que_id_file_name, sizeof(MessageQue::QuePool));
    MessageQue::QuePool* qpol = (MessageQue::QuePool*) crntmq.get(PROT_READ);

    std::stringstream ss;
    ss << std::endl << que_id_file_name << std::endl;
    int cnt = 0;
    for (int i = 0; i < (sizeof(MessageQue::QuePool)
            / sizeof(MessageQue::QueRecord)); i++) {
        //        if (bl || 0 < qpol->ques[i].send_count || 0 < qpol->ques[i].wait_count) {
        if (bl || 0 < std::string(qpol->ques[i].queName).length()) {
            ss << "    ";
            if (i < 10)
                ss << '0';
            ss << i << ' ' << qpol->ques[i].queName << " send:"
                    << qpol->ques[i].send_count << " wait:"
                    << qpol->ques[i].wait_count << std::endl;
            cnt++;
        }
    }
    ss << "    display count:" << cnt << std::endl;
    crntmq.end();

    return ss.str();
}

}
}
